package com.shujia;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;

/**
 * @author yangjiming
 * @create 2021-04-21 23:15
 */
public class Demo02API {
    private Connection conn;
    private Admin admin;

    @Before
    public void init() {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        try {
            conn = ConnectionFactory.createConnection(conf);
            admin = conn.getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    @Test
    public void createTable() {
        String name = "tb1";
        HTableDescriptor tb1 = new HTableDescriptor(TableName.valueOf(name));
        HColumnDescriptor cf1 = new HColumnDescriptor("info");
        tb1.addFamily(cf1);
        try {
            admin.createTable(tb1);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void put() throws IOException {
        Table tb1 = conn.getTable(TableName.valueOf("tb1"));
        Put put = new Put("000001".getBytes());

        put.addColumn("info".getBytes(), "name".getBytes(), "zhangsan".getBytes());
        tb1.put(put);

    }

    @Test
    public void get() throws IOException {
        Table tb1 = conn.getTable(TableName.valueOf("tb1"));
        Get get = new Get("000001".getBytes());
        Result rs = tb1.get(get);
        byte[] value = rs.getValue("info".getBytes(), "name".getBytes());
        System.out.println(Bytes.toString(value));

    }

    @Test
    public void scan() throws IOException {
        Table tb1 = conn.getTable(TableName.valueOf("tb1"));
        Scan scan = new Scan();
//        scan.withStartRow("row1".getBytes());
//        scan.withStopRow("row3".getBytes());

        ResultScanner scanner = tb1.getScanner(scan);
        Result rs;
        while ((rs = scanner.next()) != null) {
            byte[] value = rs.getValue("info".getBytes(), "name".getBytes());
            System.out.println(Bytes.toString(value));
        }
    }

    @Test
    public void putAll() throws IOException {
        //先判断表是否存在，不存在则创建
        if (!admin.tableExists(TableName.valueOf("students"))) {
            HTableDescriptor students = new HTableDescriptor(TableName.valueOf("students"));
            students.addFamily(new HColumnDescriptor("info"));
            admin.createTable(students);
        }

        Table students = conn.getTable(TableName.valueOf("students"));
        BufferedReader br = new BufferedReader(new FileReader("data/students.txt"));
        String line;
        while ((line = br.readLine()) != null) {
            String[] splits = line.split(",");
            String id = splits[0];
            String name = splits[1];
            int age = Integer.parseInt(splits[2]);
            String gender = splits[3];
            String clazz = splits[4];

            Put put = new Put(id.getBytes());

            put.addColumn("info".getBytes(), "name".getBytes(), name.getBytes());
            put.addColumn("info".getBytes(), "age".getBytes(), Bytes.toBytes(age));
            put.addColumn("info".getBytes(), "gender".getBytes(), gender.getBytes());
            put.addColumn("info".getBytes(), "clazz".getBytes(), clazz.getBytes());

            students.put(put);
        }
    }

    @Test
    public void cells() throws IOException {
        Table students = conn.getTable(TableName.valueOf("students"));
        Scan scan = new Scan();

        scan.withStartRow("1500100001".getBytes());
        scan.withStopRow("1500100021".getBytes());

        ResultScanner scanner = students.getScanner(scan);

        for (Result rs : scanner) {
            byte[] row = rs.getRow();
            String rowkey = Bytes.toString(row);
            System.out.println("学号：" + rowkey);

            for (Cell cell : rs.listCells()) {
                String family = Bytes.toString(CellUtil.cloneFamily(cell));
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                byte[] bytes = CellUtil.cloneValue(cell);
                String value;
                if ("age".equals(qualifier)) {
                    value = String.valueOf(Bytes.toInt(bytes));
                } else {
                    value = Bytes.toString(bytes);
                }

                System.out.println(qualifier + "：" + value);
            }

        }

    }

    @Test
    public void dropTable() {
        String name = "dianxin";
        try {
            admin.disableTable(TableName.valueOf(name));
            admin.deleteTable(TableName.valueOf(name));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @After
    public void close() {
        try {
            if (conn != null) {
                conn.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
